package com.smartboot.plus.features.cluster;

import com.alibaba.fastjson2.JSON;
import com.smartboot.plus.condition.ConditionGroup;
import com.smartboot.plus.condition.ConditionQ;
import com.smartboot.plus.enums.BrokerClusterEnum;
import com.smartboot.plus.enums.BrokerNodeTypeEnum;
import com.smartboot.plus.enums.BrokerStatueEnum;
import com.smartboot.plus.features.NodeUtil;
import com.smartboot.plus.holder.BrokerContextHolder;
import com.smartboot.plus.ignite.cache.MqttNodeCache;
import com.smartboot.plus.plugins.PluginConfig;
import com.smartboot.plus.service.MqttNodeService;
import com.smartboot.plus.utils.Func;
import com.smartboot.plus.utils.INetUtil;
import org.apache.ignite.Ignite;
import tech.smartboot.feat.cloud.annotation.Autowired;
import tech.smartboot.feat.cloud.annotation.Bean;
import tech.smartboot.feat.cloud.annotation.PostConstruct;
import tech.smartboot.feat.cloud.annotation.PreDestroy;
import tech.smartboot.feat.core.common.logging.Logger;
import tech.smartboot.feat.core.common.logging.LoggerFactory;
import tech.smartboot.mqtt.client.MqttClient;
import tech.smartboot.mqtt.common.util.ValidateUtils;
import tech.smartboot.mqtt.plugin.spec.BrokerContext;
import tech.smartboot.mqtt.plugin.spec.Message;
import tech.smartboot.mqtt.plugin.spec.bus.EventType;
import tech.smartboot.mqtt.plugin.spec.bus.MessageBusConsumer;

import java.sql.Timestamp;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Bean
public class ClusterFeatures {

    private static final Logger LOGGER = LoggerFactory.getLogger(ClusterFeatures.class);


    @Autowired
    private BrokerContext brokerContext;

    @Autowired
    private PluginConfig pluginConfig;

    @Autowired
    private Ignite igniteInstance;

    @Autowired
    private MqttNodeService mqttNodeService;

    @Autowired
    private ClusterConnector clusterConnector;

    private String localNodeId;

    @PostConstruct
    public void init() {

        try {

            initLocalNode();
            startNodeStatusMonitor();
            subscribeCoordinationEventBus();
            LOGGER.info("ClusterFeatures init success");
        } catch (Exception e) {
            LOGGER.error("ClusterFeatures init failed", e);
            throw new RuntimeException(e);
        }

//
//        HashedWheelTimer.DEFAULT_TIMER.scheduleWithFixedDelay(this::loadClusterNode, 5L, TimeUnit.SECONDS);
        // initCoreNode(brokerNode);
    }


    /**
     * 初始化本地节点信息，写入 Ignite 缓存
     */
    private void initLocalNode() {

        ValidateUtils.notNull(brokerContext.Options(), "brokerConfigure is null");

        localNodeId = NodeUtil.getNodeId(brokerContext);
        ValidateUtils.notBlank(localNodeId, "broker.nodeId is null");


        MqttNodeCache mqttNodeCache = new MqttNodeCache();

        mqttNodeCache.setNodeId(localNodeId);
        String host = brokerContext.Options().getHost();
        int port = brokerContext.Options().getPort();
        if ("0.0.0.0".equals(host) || host.isEmpty()) {
            host = INetUtil.getHostIp();
        }

        mqttNodeCache.setIpAddress(host);
        mqttNodeCache.setClusterEndpoint("mqtt://" + host + ":" + port);
        mqttNodeCache.setPort(port);

        mqttNodeCache.setNodeClientId(MqttClientIdGenerator.generate());
        mqttNodeCache.setClusterPassword(Func.randomUUID());
        mqttNodeCache.setClusterPassword(Func.randomUUID());

        Timestamp now = new Timestamp(System.currentTimeMillis());
        mqttNodeCache.setUpAt(now);
        mqttNodeCache.setUpdatedAt(now);
        //节点类型core
        mqttNodeCache.setNodeType(BrokerNodeTypeEnum.CORE_NODE.getCode());
        //节点状态 运行中
        mqttNodeCache.setStatus(BrokerStatueEnum.RUNNING.getCode());
        //集群开关 0集群，1不集群
        mqttNodeCache.setClusterSwitch(BrokerClusterEnum.NO.getCode());
        //默认集群状态停止
        mqttNodeCache.setClusterStatus(BrokerStatueEnum.STOPPED.getCode());

        mqttNodeService.getIgniteCache().put(localNodeId, mqttNodeCache);

        LOGGER.info("Local node [{}] initialized: {}", localNodeId, JSON.toJSONString(mqttNodeCache));
    }


    /**
     * 启动集群节点状态同步定时任务（每5秒）
     * 负责从 Ignite 中加载所有节点状态到内存缓存
     * 负责建立或断开与远程节点的连接
     */
    private void startNodeStatusMonitor() {
        syncClusterNodesStatus();
        this.brokerContext.getTimer().scheduleWithFixedDelay(() -> {
            try {
                syncClusterNodesStatus();
                manageRemoteNodeConnections();
            } catch (Exception e) {
                LOGGER.error("Error syncing cluster nodes status", e);
            }
        }, 5L, TimeUnit.SECONDS);
        LOGGER.info("Node status monitor started");
    }

    /**
     * 从 Ignite 中读取所有节点，更新本地缓存状态
     */
    private void syncClusterNodesStatus() {
        List<MqttNodeCache> nodeList = mqttNodeService.selectList(new ConditionGroup(), MqttNodeCache.class);
        if (Func.isEmpty(nodeList)) {
            return;
        }

        Map<String, MqttNodeCache> cacheMap = BrokerContextHolder.getMqttNodeCacheMap();

        for (MqttNodeCache node : nodeList) {
            //if (BrokerStatueEnum.RUNNING.getCode().equals(node.getStatus())) {
            cacheMap.put(node.getNodeId(), node);
            // }
//            else {
//                cacheMap.remove(node.getNodeId());
//            }
        }

    }


    /**
     * 管理集群间节点连接
     */
    private void manageRemoteNodeConnections() {
        MqttNodeCache localNode = mqttNodeService.selectOne(new ConditionGroup().add(new ConditionQ(MqttNodeCache::getNodeId, localNodeId)), MqttNodeCache.class);
        if (localNode == null) {
            LOGGER.warn("Local node [{}] not found in cache", localNodeId);
            return;
        }

        // 遍历所有远程节点，判断是否需要连接或断开
        ConditionGroup conditionGroup = new ConditionGroup().add(new ConditionQ(MqttNodeCache::getNodeId, ConditionQ.NOT_EQ, localNodeId));
        List<MqttNodeCache> clusterNodes = mqttNodeService.selectList(conditionGroup, MqttNodeCache.class);

        List<String> nodeIds = clusterNodes.stream().map(MqttNodeCache::getNodeId).collect(Collectors.toList());
        //移除不存在的集群节点
        BrokerContextHolder.getConnectedNodeMap().keySet().stream().filter(nodeId -> !nodeIds.contains(nodeId)).forEach(nodeId -> {
            BrokerContextHolder.getConnectedNodeMap().remove(nodeId).disconnect();
        });

        for (MqttNodeCache remoteNode : clusterNodes) {
            String remoteNodeId = remoteNode.getNodeId();

            /**
             * 判断本地是否已与该远程节点建立连接
             */
            boolean connectionAlreadyExists = BrokerContextHolder.getConnectedNodeMap().containsKey(remoteNodeId);

            /**
             * 如果远程节点不应连接，但当前已连接，断开连接
             */
            if (remoteNode.getClusterSwitch() == BrokerClusterEnum.NO.getCode() && connectionAlreadyExists) {
                System.out.println("但当前已连接，断开连接");
                disconnectFromNode(remoteNode);
                continue;
            }
            /**
             * 如果远程节点应连接，但尚未连接，发起连接
             */
            if (remoteNode.getClusterSwitch() == BrokerClusterEnum.YES.getCode() && !connectionAlreadyExists) {
                System.out.println("但尚未连接，发起连接");
                clusterConnector.connect(remoteNode);
            }
        }
    }

    /**
     * 断开与某远程节点的连接，更新状态
     */
    private void disconnectFromNode(MqttNodeCache remoteNode) {
        String nodeId = remoteNode.getNodeId();
        MqttClient client = BrokerContextHolder.getConnectedNodeMap().remove(nodeId);
        if (client != null && client.isDisconnect()) {
            client.disconnect();
            LOGGER.warn("Disconnected from node [{}]", nodeId);
        }
        // 更新远程节点状态为 STOPPED
        remoteNode.setClusterStatus(BrokerStatueEnum.STOPPED.getCode());
        remoteNode.setUpAt(new Timestamp(System.currentTimeMillis()));
        mqttNodeService.getIgniteCache().put(nodeId, remoteNode);
    }

    /**
     * 订阅集群间协调事件（连接、断开、消息转发）
     */
    private void subscribeCoordinationEventBus() {
        ConditionGroup conditionGroup = new ConditionGroup().add(new ConditionQ(MqttNodeCache::getNodeId, localNodeId));
        MqttNodeCache localNode = mqttNodeService.selectOne(conditionGroup, MqttNodeCache.class);
        if (localNode == null) {
            LOGGER.warn("Local node not found for event subscription");
            return;
        }

        //处理断开事件，清理认证会话映射
        brokerContext.getEventBus().subscribe(EventType.DISCONNECT, (type, session) -> {
            BrokerContextHolder.getCoreSessionMap().remove(session);
        });


        // 获取所有连接的节点
        Map<String, MqttClient> connectedNodes = BrokerContextHolder.getConnectedNodeMap();
        //转发到所有已连接节点
        brokerContext.getMessageBus().consumer((session, message) -> {

            // 遍历所有连接的节点
            for (Map.Entry<String, MqttClient> entry : connectedNodes.entrySet()) {
                String nodeId = entry.getKey();
                MqttClient client = entry.getValue();

                // 如果当前节点是本地节点，跳过
                if (nodeId.equals(localNodeId)) {
                    continue;
                }
                String topic = message.getTopic().getTopic();
                // 发布消息到其他节点
                client.publish(topic, message.getQos(), message.getPayload(), message.isRetained(), ack -> {
                });
            }
            //   LOGGER.debug("Message dispatched from session [{}] topic [{}] to all connected nodes", session.getClientId(), topic);
        });
    }

    @PreDestroy
    public void destroy() {

        LOGGER.warn("Node [{}] shutting down", localNodeId);
        MqttNodeCache node = mqttNodeService.getIgniteCache().get(localNodeId);
        if (node != null) {
            node.setStatus(BrokerStatueEnum.STOPPED.getCode());
            node.setUpdatedAt(new Timestamp(System.currentTimeMillis()));
            mqttNodeService.getIgniteCache().put(localNodeId, node);
        }
    }
}
